package org.example.log;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class DataFromCustom {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        并行度1
//        DataStreamSource<EventCount> single = env.addSource(new MySource());
//        single.print();
//        并行度》1
        DataStreamSource<EventCount> partial = env.addSource(new ParalleSourece())
                .setParallelism(5);
        partial.print();

        env.execute();

    }

    /**
     * 多并行度数据源
     */
    public static class ParalleSourece implements ParallelSourceFunction<EventCount> {
        private boolean running = true;

        @Override
        public void run(SourceContext<EventCount> ctx) throws Exception {
            while (running) {
                EventCount eventCount = new EventCount(UUID.randomUUID().toString(), UUID.randomUUID().toString(), System.currentTimeMillis());
                ctx.collect(eventCount);
                TimeUnit.SECONDS.sleep(1);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }


    }

    /**
     * 单并行度数据源
     */
    public static class MySource implements SourceFunction<EventCount> {
        private boolean running = true;

        @Override
        public void run(SourceContext<EventCount> ctx) throws Exception {
            while (running) {
                EventCount eventCount = new EventCount(UUID.randomUUID().toString(), UUID.randomUUID().toString(), System.currentTimeMillis());
                ctx.collect(eventCount);
                TimeUnit.SECONDS.sleep(1);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}